package com.lds.memory.kafka.producer;

import com.lds.memory.kafka.partition.Partition;
import com.lds.memory.kafka.partition.selector.PartitionSelectorFactory;
import com.lds.memory.kafka.topic.Topic;

import java.util.Collection;
import java.util.List;

public class Producer<T> implements IProducer<T> {

    private final Topic topic;

    public Producer(Topic topic) {
        this.topic = topic;
    }

    @Override
    public void send(T data) {
        List<Partition> partitions = topic.getPartitions();
        int partitionIndex = PartitionSelectorFactory.getDefault().select(partitions.size(), null);
        Partition partition = partitions.get(partitionIndex);
        partition.store(data);
    }

    @Override
    public void send(T data, String partitionKey) {
        List<Partition> partitions = topic.getPartitions();
        int partitionIndex = PartitionSelectorFactory.getHash().select(partitions.size(), partitionKey);
        Partition partition = partitions.get(partitionIndex);
        partition.store(data);
    }

    @Override
    public void batchSend(Collection<T> dataList) {
        for (T data : dataList) {
            send(data);
        }
    }

    @Override
    public void batchSend(Collection<T> dataList, String partitionKey) {
        for (T data : dataList) {
            send(data, partitionKey);
        }
    }

}
